 1.Kafka高级特性解析之消费者中消息接收
   
   1).必要参数配置      
   参数                说明
   bootstrap.servers   向Kafka集群建立初始连接用到的host/port列表。
                       客户端会使用这里列出的所有服务器进行集群其他服务器的发现，而不管是否指定了哪个服务器用作引导。
                       这个列表仅影响用来发现集群所有服务器的初始主机。
                       字符串形式：host1:port1,host2:port2,...
                       由于这组服务器仅用于建立初始链接，然后发现集群中的所有服务器，因此没有必要将集群中的所有地址写在这里。
                       一般最好两台，以防其中一台宕掉。
   key.deserializer    key的反序列化类，该类需要实现org.apache.kafka.common.serialization.Deserializer 接口。
   value.deserializer  实现了org.apache.kafka.common.serialization.Deserializer 接⼝的反序列化器，
                       用于对消息的value进行反序列化。
   client.id           当从服务器消费消息的时候向服务器发送的id字符串。在ip/port基础上提供应用的逻辑名称，
                       记录在服务端的请求日志中，用于追踪请求的源。
   group.id            用于唯一标志当前消费者所属的消费组的字符串。
                       如果消费者使用组管理功能如subscribe(topic)或使用基于Kafka的偏移量管理策略，该项必须设置。   
   auto.offset.reset   当Kafka中没有初始偏移量或当前偏移量在服务器中不存在（如，数据被删除了），该如何处理？
                       earliest：自动重置偏移量到最早的偏移量
                       latest：自动重置偏移量为最新的偏移量
                       none：如果消费组原来的（previous）偏移量不存在，则向消费者抛异常
                       anything：向消费者抛异常
   enable.auto.commit  如果设置为true，消费者会自动周期性地向服务器提交偏移量。
   2).订阅
   Topic，Kafka用于分类管理消息的逻辑单元，类似与MySQL的数据库。
   Partition，是Kafka下数据存储的基本单元，这个是物理上的概念。同一个topic的数据，会被分散的存储
到多个partition中，这些partition可以在同一台机器上，也可以是在多台机器上。优势在于：有利于水平扩
展，避免单台机器在磁盘空间和性能上的限制，同时可以通过复制来增加数据冗余性，提高容灾能力。为了做
到均匀分布，通常partition的数量通常是Broker Server数量的整数倍。   
   Consumer Group，同样是逻辑上的概念，是Kafka实现单播和广播两种消息模型的手段。保证一个消费组
获取到特定主题的全部的消息。在消费组内部，若干个消费者消费主题分区的消息，消费组可以保证一个主
题的每个分区只被消费组中的一个消费者消费。
   consumer 采用 pull 模式从 broker 中读取数据。
   采用 pull 模式，consumer 可自主控制消费消息的速率， 可以自己控制消费方式（批量消费/逐条消费)，还可以
选择不同的提交方式从而实现不同的传输语义。
   consumer.subscribe("tp_demo_01,tp_demo_02")
   3).反序列化
   Kafka的broker中所有的消息都是字节数组，消费者获取到消息之后，需要先对消息进行反序列化处理，然后才能
交给用户程序消费处理。
   消费者的反序列化器包括key的和value的反序列化器。
   key.deserializer
   value.deserializer
   IntegerDeserializer
   StringDeserializer
   需要实现org.apache.kafka.common.serialization.Deserializer<T> 接口。
   消费者从订阅的主题拉取消息：
   consumer.poll(3_000);
   在Fetcher类中，对拉取到的消息首先进行反序列化处理。
   Kafka默认提供了几个反序列化的实现：
   org.apache.kafka.common.serialization 包下包含了这几个实现：
   包含 ByteArrayDeserializer、ByteBufferDeserializer、ByteDeserializer、DoubleDeserializer、FloatDeserializer
   IntegerDeserializer、LongDeserializer、ShortDeserializer及StringDeserializer类
   (1).自定义反序列化
   自定义反序列化类，需要实现org.apache.kafka.common.serialization.Deserializer<T> 接口。
   com.lagou.kafka.demo.entity.User
package com.lagou.kafka.demo.entity;

/**
 * 用户自定义的封装消息的实体类
 */
public class User {
    private Integer userId;
    private String username;

    public User() {
    }

    public User(Integer userId, String username) {
        this.userId = userId;
        this.username = username;
    }

    public Integer getUserId() {
        return userId;
    }

    public void setUserId(Integer userId) {
        this.userId = userId;
    }

    public String getUsername() {
        return username;
    }

    public void setUsername(String username) {
        this.username = username;
    }

    @Override
    public String toString() {
        return "User{" +
                "userId=" + userId +
                ", username='" + username + '\'' +
                '}';
    }
}

   com.lagou.kafka.demo.deserializer.UserDeserializer
package com.lagou.kafka.demo.deserializer;

import com.lagou.kafka.demo.entity.User;
import org.apache.kafka.common.serialization.Deserializer;

import java.nio.ByteBuffer;
import java.util.Map;

public class UserDeserializer implements Deserializer<User> {
    @Override
    public void configure(Map<String, ?> map, boolean b) {

    }

    @Override
    public User deserialize(String topic, byte[] data) {
        ByteBuffer buffer = ByteBuffer.allocate(data.length);
        buffer.put(data);
        buffer.flip();
        final int userId = buffer.getInt();
        final int usernameLength = buffer.getInt();
        String username = new String(data, 8 , usernameLength);

        return new User(userId, username);
    }

    @Override
    public void close() {

    }
}

   com.lagou.kafka.demo.consumer.UserConsumer
package com.lagou.kafka.demo.consumer;

import com.lagou.kafka.demo.deserializer.UserDeserializer;
import com.lagou.kafka.demo.entity.User;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

public class UserConsumer {

    public static void main(String[] args) {
        Map<String, Object> configs = new HashMap<String, Object>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "linux121:9092");
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, UserDeserializer.class);
        // 设置自定义的反序列化器
        configs.put(ConsumerConfig.GROUP_ID_CONFIG, "user_consumer");
        configs.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer_id");
        configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        KafkaConsumer<String, User> consumer = new KafkaConsumer<String, User>(configs);
        //订阅主题
        consumer.subscribe(Collections.singleton("tp_user_01"));
        final ConsumerRecords<String, User> records = consumer.poll(Long.MAX_VALUE);
        records.forEach(new Consumer<ConsumerRecord<String, User>>() {
            @Override
            public void accept(ConsumerRecord<String, User> record) {
                System.out.println(record.value());
            }
        });

        //关闭消费者
        consumer.close();
    }
}
s